package business

import (
	"context"
	"golang-manage-admin/queue/publish"
	"time"

	"github.com/cihub/seelog"
	"github.com/streadway/amqp"
	"golang-manage-admin/providers"
)

/**
 * 队列接口
 */
type IQueue interface {
	GetConfigKey() string                            // 获取配置queue 下的 key
	SetLogger(logger seelog.LoggerInterface)         // 设置日志
	GetLogger() seelog.LoggerInterface               // 获取日志
	Delivery(ctx context.Context, msg amqp.Delivery) // 接收的消息响应
	EnsureMsg(msg amqp.Delivery, mode ResponseMode)  // 确定消息,由于外网网络环境较差，可能丢包等情况，需要重新确定消息
	Init()                                           // 初始化
	Close()                                          // 关闭清理
}

/**
 * 等待确定元素结构
 */
type WaitQueue struct {
	msg    amqp.Delivery
	action ResponseMode
}

/**
 * 消息应答结构
 */
type ResponseMode struct {
	ResponseType   string      // ack : 消费消息； reject ：拒绝消费； retry ：推送死信重试
	ToPublishKey   int         // ack: 模式下不需要 ; reject 和 retry 模式需要转发（目标）
	FromPublishKey int         // ack: 模式下不需要 ; reject 和 retry 模式需要转发（来自）
	Data           interface{} // ack: 模式下不需要 ; reject 和 retry 模式需要转发（内容体）
}

/**
 * 队列接口的通用实现
 */
type Implementation struct {
	logger      seelog.LoggerInterface
	stockLogger seelog.LoggerInterface

	// @TODO 这里其实就算是被重启清理掉也没关系，因为这里是为了重试所用，而程序中也有去重处理,不存在丢失消息的情况
	waitQueues chan WaitQueue
}

func (i *Implementation) SetLogger(logger seelog.LoggerInterface) {
	i.logger = logger
}

func (i *Implementation) GetLogger() seelog.LoggerInterface {
	return i.logger
}

func (i *Implementation) Delivery(msg *amqp.Delivery) {
}

func (i *Implementation) EnsureMsg(msg amqp.Delivery, mode ResponseMode) {
	if mode.ResponseType != "ack" && mode.ResponseType != "reject" && mode.ResponseType != "retry" {
		return
	}

	i.waitQueues <- WaitQueue{
		msg:    msg,
		action: mode,
	}
}

func (i *Implementation) waitQueueHandler() {
	for waitQueue := range i.waitQueues {
		// 正常响应
		if waitQueue.action.ResponseType == "ack" {
			err := waitQueue.msg.Ack(false)
			if err != nil {
				i.EnsureMsg(waitQueue.msg, waitQueue.action)
				time.Sleep(5 * time.Second)
			}
			continue
		}

		// 拒绝消息延迟一秒执行
		if waitQueue.action.ResponseType == "reject" {
			rejectWaitQueue := waitQueue
			time.AfterFunc(
				time.Second*1,
				func() {
					err := rejectWaitQueue.msg.Reject(true)
					if err != nil {
						// @Todo 判断重试次数
						time.Sleep(1 * time.Second)
						i.EnsureMsg(rejectWaitQueue.msg, rejectWaitQueue.action)
					}
				},
			)
			continue
		}

		// 重试消息--延迟一秒执行
		if waitQueue.action.ResponseType == "retry" {
			// @Todo 推送到死信队列
			sendErr := publish.PublishObject.PushFormRetryMessage(waitQueue.action.ToPublishKey, waitQueue.action.FromPublishKey, waitQueue.msg, waitQueue.action.Data)
			if sendErr != nil {
				time.AfterFunc(
					time.Second*1,
					func() {
						//  @Todo 判断重试次数
						time.Sleep(1 * time.Second)
						i.EnsureMsg(waitQueue.msg, waitQueue.action)
					},
				)
			} else {
				// @Todo 死信推送完成，关闭原消息
				err := waitQueue.msg.Ack(false)
				if err != nil {
					// 改成正常ack；不能重复推送死信
					waitQueue.action.ResponseType = "ack"
					i.EnsureMsg(waitQueue.msg, waitQueue.action)
					time.Sleep(5 * time.Second)
				}
			}
			continue
		}
	}
}

func (i *Implementation) Init() {
	i.waitQueues = make(chan WaitQueue, providers.Config.GetInt("app.wait_queue_size"))

	// 开始监听
	go i.waitQueueHandler()
}

func (i *Implementation) Close() {
	close(i.waitQueues)
	i.waitQueues = nil
}

func (i *Implementation) ReplyQueueHandler(responseMode ResponseMode, msg amqp.Delivery) {
	switch responseMode.ResponseType {
	case "ack":
		err := msg.Ack(false)
		if err != nil {
			_ = i.logger.Errorf("消息处理确定失败:%s---%s", err.Error(), string(msg.Body))
			i.EnsureMsg(msg, responseMode)
		}
		return
	case "reject":
		err := msg.Reject(true)
		if err != nil {
			_ = i.logger.Errorf("消息处理拒绝失败:%s---%s", err.Error(), string(msg.Body))
			i.EnsureMsg(msg, responseMode)
		}
		return
	case "retry":
		// @Todo 消息推送死信队列
		sendErr := publish.PublishObject.PushFormRetryMessage(responseMode.ToPublishKey, responseMode.FromPublishKey, msg, responseMode.Data)
		if sendErr != nil {
			_ = i.logger.Error("消息处理推送死信队列失败:%s---%s", sendErr.Error(), string(msg.Body))
			i.EnsureMsg(msg, responseMode)
			return
		}

		// @Todo 关闭原消息
		err := msg.Ack(false)
		if err != nil {
			_ = i.logger.Errorf("消息处理确定失败:%s---%s", err.Error(), string(msg.Body))
			i.EnsureMsg(msg, ResponseMode{
				ResponseType: "ack",
			})
		}
		return
	default:
		_ = i.logger.Errorf("消息处理类型有误:%s", responseMode.ResponseType)
		return
	}

}
